package com.lvmama.java.rhino.etl.runner;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;

import org.apache.commons.collections.CollectionUtils;
import org.apache.log4j.Logger;
import org.apache.spark.sql.SQLContext;
import org.springframework.beans.BeanUtils;

import com.alibaba.fastjson.JSON;
import com.lvmama.java.rhino.spark.entity.AccessLog;
import com.lvmama.java.rhino.spark.entity.BaseLog;
import com.lvmama.java.rhino.spark.entity.DeviceInfoLog;
import com.lvmama.java.rhino.spark.entity.LogDetailInfo;
import com.lvmama.java.rhino.spark.entity.RequestTimeInfo;
import com.lvmama.java.rhino.spark.entity.UserInfoLog;
import com.lvmama.java.rhino.spark.tables.ElasticLogBaseParam;
import com.lvmama.java.rhino.spark.tables.ElasticLogDetail;
import com.lvmama.java.rhino.spark.tables.ElasticRequestTimeDetail;
import com.lvmama.java.rhino.spark.tables.LogDetail;
import com.lvmama.java.rhino.spark.tables.RequestSpecialParam;
import com.lvmama.java.rhino.spark.tables.RequestTimeDetail;
import com.lvmama.java.rhino.spark.tables.UserAccessPath;
import com.lvmama.java.rhino.spark.utils.Constants;
import com.lvmama.java.rhino.spark.utils.ElasticUtils;
import com.lvmama.java.rhino.spark.utils.LogProcessorUtil;
import com.lvmama.java.rhino.spark.utils.SaveToHdfsUtils;

import kafka.consumer.KafkaStream;
import kafka.message.MessageAndMetadata;

public class SaveLogDataRunner {
	private static final Logger LOGGER = Logger.getLogger(SaveLogDataRunner.class);
	private SQLContext sqlContext;
	private KafkaStream<byte[], byte[]> kafkaStream;
	public SaveLogDataRunner(SQLContext sqlContext, KafkaStream<byte[], byte[]> kafkaStream) {
		this.sqlContext = sqlContext;
		this.kafkaStream = kafkaStream;
	}
	
	public void run() {
		List<LogDetailInfo> dataList = new ArrayList<LogDetailInfo>();
		for (MessageAndMetadata<byte[], byte[]> mm : kafkaStream) {
			byte[] rhinoMessage = mm.message();
			// 得到对象类型消息,做逻辑处理
			try {
				Object obj = deserializeRovinceCities(new String(rhinoMessage, "UTF-8"));
				if(!(obj instanceof LogDetailInfo)) {
					continue;
				}
				LogDetailInfo logDetailInfo = (LogDetailInfo) obj;
				dataList.add(logDetailInfo);
				//if(dataList.size() >= 3) {
					saveLogData(dataList);
					dataList.clear();
				//}
			} catch (Exception e) {
				LOGGER.error(e.getMessage(), e);
			}
		}
	}

	private void saveLogData(List<LogDetailInfo> dataList) {
		String realTimeSyncElastic = Constants.getInstance().getValue("elastic.realtime.synchronous");
		if("true".equals(realTimeSyncElastic)) {
			for (LogDetailInfo data : dataList) {
				saveToElasticSearch(data);
			}
		}
		saveToHdfs(dataList);
	}
	
	private void saveToHdfs(List<LogDetailInfo> parseResultList) {
		List<LogDetail> logDetailList = new ArrayList<LogDetail>();
		List<UserAccessPath> userAccessPathList = new ArrayList<UserAccessPath>();
		List<RequestTimeDetail> apiRequestTimeDetailList = new ArrayList<RequestTimeDetail>();
		List<RequestSpecialParam> requestSpecialParamList = new ArrayList<RequestSpecialParam>();
		
		for(LogDetailInfo parseResult : parseResultList) {
			LogDetail logDetail = new LogDetail();
			logDetail.setThreadName(parseResult.getThreadName());
//			logDetail.setContent(LogProcessorUtil.gzip(LogProcessorUtil.concatLog(parseResult.getLogList())));
			logDetail.setContent("测试短的能HDFS吗");
			logDetailList.add(logDetail);
			
			UserAccessPath userAccessPath = combUserAccessPath(parseResult);
			userAccessPathList.add(userAccessPath);
			
			List<RequestTimeDetail> tempApiRequestTimeDetailList = parseResult.getApiRequestTime();
			apiRequestTimeDetailList.addAll(tempApiRequestTimeDetailList);
			
			RequestSpecialParam requestSpecialParam = combRequestSpecialParam(parseResult);
			if(requestSpecialParam != null) {
				requestSpecialParamList.add(requestSpecialParam);
			}
			System.out.println("正在保存日志到HDFS，ThreadName: " + parseResult.getThreadName());
			LOGGER.info("正在保存日志到，ThreadName: " + parseResult.getThreadName());
		}
		
		SaveToHdfsUtils.save(sqlContext, SaveToHdfsUtils.TABLE_NAME.LOG_DETAIL, logDetailList);
		SaveToHdfsUtils.save(sqlContext, SaveToHdfsUtils.TABLE_NAME.USER_ACCESS_PATH, userAccessPathList);
		SaveToHdfsUtils.save(sqlContext, SaveToHdfsUtils.TABLE_NAME.API_REQUEST_TIME_DETAIL, apiRequestTimeDetailList);
		SaveToHdfsUtils.save(sqlContext, SaveToHdfsUtils.TABLE_NAME.REQUEST_SPECIAL_PARAM, requestSpecialParamList);
	}
	
	private RequestSpecialParam combRequestSpecialParam(LogDetailInfo parseResult) {
		RequestSpecialParam requestSpecialParam = new RequestSpecialParam();
		List<String> specialParam = parseResult.getSpecialLog();
		if(specialParam == null) {
			return null;
		}
		for (int i = 0; i < specialParam.size(); i++) {
			if(i == 0) {
				requestSpecialParam.setParam1(specialParam.get(0));
			}
			if(i == 1) {
				requestSpecialParam.setParam2(specialParam.get(1));
			}
			if(i == 2) {
				requestSpecialParam.setParam3(specialParam.get(2));
			}
			if(i == 3) {
				requestSpecialParam.setParam4(specialParam.get(3));
			}
			if(i == 4) {
				requestSpecialParam.setParam5(specialParam.get(4));
			}
			if(i == 5) {
				requestSpecialParam.setParam6(specialParam.get(5));
			}
			if(i == 6) {
				requestSpecialParam.setParam7(specialParam.get(6));
			}
			if(i == 7) {
				requestSpecialParam.setParam8(specialParam.get(7));
			}
			if(i == 8) {
				requestSpecialParam.setParam9(specialParam.get(8));
			}
			if(i == 9) {
				requestSpecialParam.setParam10(specialParam.get(9));
			}
		}
		return requestSpecialParam;
	}
	
	private UserAccessPath combUserAccessPath(LogDetailInfo parseResult) {
		UserAccessPath userAccessPath = new UserAccessPath();
		userAccessPath.setThreadName(parseResult.getThreadName());
		BaseLog baseLog = parseResult.getBaseLog();
		if(baseLog != null) {
			if(baseLog.deviceInfo != null) {
				BeanUtils.copyProperties(baseLog.deviceInfo, userAccessPath);
			}
			if(baseLog.userInfo != null) {
				BeanUtils.copyProperties(baseLog.userInfo, userAccessPath);
			}
			if(baseLog.accessLog != null) {
				BeanUtils.copyProperties(baseLog.accessLog, userAccessPath);
			}
		}
		RequestTimeInfo requestTimeInfo = parseResult.getRequestTimeInfo();
		if(requestTimeInfo != null) {
			userAccessPath.setRequestTime(requestTimeInfo.getRequestTime());
			userAccessPath.setResponseTime(requestTimeInfo.getResponseTime());
		}
		return userAccessPath;
	}
	
	private void saveToElasticSearch(LogDetailInfo parseResult) {
		ElasticLogBaseParam logBaseParam = combLogBaseParam(parseResult);
		ElasticLogDetail elasticLogDetail = new ElasticLogDetail();
		elasticLogDetail.setThreadName(parseResult.getThreadName());
		elasticLogDetail.setLogDetail(parseResult.getLogList());
		try {
			ElasticUtils.getInstance().putIndex(Constants.ELASTIC_INDEX_LVTU_INDEX, Constants.ELASTIC_TYPE_BASE_PARAM_TYPE, JSON.toJSONString(logBaseParam));
			ElasticUtils.getInstance().putIndex(Constants.ELASTIC_INDEX_LVTU_INDEX, Constants.ELASTIC_TYPE_LOG_DETAIL_TYPE, JSON.toJSONString(elasticLogDetail));
			System.out.println("日志存储Elasticsearch完成，ThreadName: " + parseResult.getThreadName());
			LOGGER.info("日志存储Elasticsearch完成，ThreadName: " + parseResult.getThreadName());
		} catch (InterruptedException e) {
			LOGGER.error(e.getMessage(), e);
		} catch (ExecutionException e) {
			LOGGER.error(e.getMessage(), e);
		}
	}

	private ElasticLogBaseParam combLogBaseParam(LogDetailInfo parseResult) {
		ElasticLogBaseParam elasticLogBaseParam = new ElasticLogBaseParam();
		elasticLogBaseParam.setThreadName(parseResult.getThreadName());
		BaseLog baseLog = parseResult.getBaseLog();
		if(baseLog != null) {
			DeviceInfoLog deviceInfoLog = baseLog.deviceInfo;
			if(deviceInfoLog != null) {
				BeanUtils.copyProperties(deviceInfoLog, elasticLogBaseParam);
			}
			UserInfoLog userInfoLog = baseLog.userInfo;
			if(userInfoLog != null) {
				BeanUtils.copyProperties(userInfoLog, elasticLogBaseParam);
			}
			AccessLog accessLog = baseLog.accessLog;
			if(accessLog != null) {
				BeanUtils.copyProperties(accessLog, elasticLogBaseParam);
			}
		}
		RequestTimeInfo requestTimeInfo = parseResult.getRequestTimeInfo();
		if(requestTimeInfo != null) {
			BeanUtils.copyProperties(requestTimeInfo, elasticLogBaseParam);
		}
		elasticLogBaseParam.setLogDetail(parseResult.getLogList());
		List<String> specialLog = parseResult.getSpecialLog();
		if(specialLog != null) {
			if(specialLog.size() > 0) {
				elasticLogBaseParam.setParam1(specialLog.get(0));
			}
			if(specialLog.size() > 1) {
				elasticLogBaseParam.setParam2(specialLog.get(1));
			}
			if(specialLog.size() > 2) {
				elasticLogBaseParam.setParam3(specialLog.get(2));
			}
			if(specialLog.size() > 3) {
				elasticLogBaseParam.setParam4(specialLog.get(3));
			}
			if(specialLog.size() > 4) {
				elasticLogBaseParam.setParam5(specialLog.get(4));
			}
			if(specialLog.size() > 5) {
				elasticLogBaseParam.setParam6(specialLog.get(5));
			}
			if(specialLog.size() > 6) {
				elasticLogBaseParam.setParam7(specialLog.get(6));
			}
			if(specialLog.size() > 7) {
				elasticLogBaseParam.setParam8(specialLog.get(7));
			}
			if(specialLog.size() > 8) {
				elasticLogBaseParam.setParam9(specialLog.get(8));
			}
			if(specialLog.size() > 9) {
				elasticLogBaseParam.setParam10(specialLog.get(9));
			}
		}
		List<RequestTimeDetail> apiRequestTimeList = parseResult.getApiRequestTime();
		if(CollectionUtils.isNotEmpty(apiRequestTimeList)) {
			List<ElasticRequestTimeDetail> requestTimeDetail = new ArrayList<ElasticRequestTimeDetail>();
			for(RequestTimeDetail apiRequestTime : apiRequestTimeList) {
				ElasticRequestTimeDetail elasticRequestTimeDetail = new ElasticRequestTimeDetail();
				BeanUtils.copyProperties(apiRequestTime, elasticRequestTimeDetail);
				requestTimeDetail.add(elasticRequestTimeDetail);
			}
			elasticLogBaseParam.setRequestTimeDetail(requestTimeDetail);
		}
		return elasticLogBaseParam;
	}
	
	/** 
	 * 将序列化字符串反序列化为集合
	 * @author wxliyong 
	 * @throws IOException 
	 * @date 2015-4-3 下午4:21:20 
	 */
	private Object deserializeRovinceCities(String value) throws Exception {
		ByteArrayInputStream byteArrayInputStream = null;
		ObjectInputStream objectInputStream = null;
		try {
			String redStr = java.net.URLDecoder.decode(value, "UTF-8");
			byteArrayInputStream = new ByteArrayInputStream(redStr.getBytes("ISO-8859-1"));
			objectInputStream = new ObjectInputStream(byteArrayInputStream);
			return objectInputStream.readObject();
		} catch (Exception e) {
			throw e;
		} finally {
			if(null != objectInputStream) {
				objectInputStream.close();
			}
			if(null != byteArrayInputStream) {
				byteArrayInputStream.close();
			}
		}
	}
}
